package com.bw.yk08;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collections;

public class MyAsyncFunction extends RichAsyncFunction<OrderInfoWide,OrderInfoWide> {

    private Connection conn1;
    private Connection conn2;
    private Connection conn3;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn1= DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
        conn2= DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
        conn3= DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
    }

    @Override
    public void close() throws Exception {
        if(conn1!=null)
            conn1.close();
        if(conn2!=null)
            conn2.close();
        if(conn3!=null)
            conn3.close();
    }

    @Override
    public void asyncInvoke(OrderInfoWide orderInfoWide, ResultFuture<OrderInfoWide> resultFuture) throws Exception {

        String sql1="select line from dim8_product_sku where id=?";
        PreparedStatement ps1=null;
        try{
            ps1=conn1.prepareStatement(sql1);
            ps1.setObject(1,orderInfoWide.getOrderDetail().getProduct_sku_id());
            java.sql.ResultSet rs1=ps1.executeQuery();
            if(rs1.next())
            {
                String line1=rs1.getString("line");
                if(line1!=null)
                {
                    JSONObject ob1= JSON.parseObject(line1);
                    Long product_spu_id=ob1.getLong("product_spu_id");
                    orderInfoWide.setProduct_spu_id(product_spu_id);
                }
            }
        }catch (Exception e)
        {
            e.printStackTrace();
        }



        String sql2="select line from dim8_shop where id=?";
        PreparedStatement ps2=null;
        try{
            ps2=conn2.prepareStatement(sql1);
            ps2.setObject(1,orderInfoWide.getOrderInfo().getShop_id());
            java.sql.ResultSet rs2=ps2.executeQuery();
            if(rs2.next())
            {
                String line1=rs2.getString("line");
                if(line1!=null)
                {
                    JSONObject ob1= JSON.parseObject(line1);
                    String region_id=ob1.getString("region_id");

                    String sql3="select line from dim8_region where id=?";
                    PreparedStatement ps3=null;
                    try{
                        ps3=conn3.prepareStatement(sql3);
                        ps3.setObject(1,region_id);
                        java.sql.ResultSet rs3=ps3.executeQuery();
                        if(rs3.next())
                        {
                            String line3=rs3.getString("line");
                            if(line3!=null)
                            {
                                JSONObject ob3= JSON.parseObject(line1);
                                String superior_region=ob1.getString("superior_region");
                                orderInfoWide.setRegion_code(superior_region);
                            }
                        }
                    }catch (Exception e)
                    {
                        e.printStackTrace();
                    }

                }
            }
        }catch (Exception e)
        {
            e.printStackTrace();
        }

        resultFuture.complete(Collections.singleton(orderInfoWide));

    }

}
